package cn.seaboot.admin.activemq;

import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

/**
 * 消息处理器
 *
 * @author Mr.css
 * @version 2024-10-15 17:20
 */
@Configuration
@IntegrationComponentScan
public class MqttClientConfig {

    // 消费端的 客户ID
    private static final String IN_CLIENT_ID = "MQTTReceiver";
    // 生产端的 客户ID
    private static final String OUT_CLIENT_ID = "MqttPublish";

    /**
     * 连接工厂，配置账号密码等信息
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();

        // mqtt-v3
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName("dev");
        options.setPassword("dev".toCharArray());
        options.setServerURIs(new String[]{"tcp://localhost:1883"});

        // 心跳包间隔，10s - 60s
        options.setKeepAliveInterval(MqttConnectOptions.KEEP_ALIVE_INTERVAL_DEFAULT);
        // 连接超时配置
        options.setConnectionTimeout(MqttConnectOptions.CONNECTION_TIMEOUT_DEFAULT);

        // 自动重连，详见：MqttAsyncClient
        options.setAutomaticReconnect(false);

        factory.setConnectionOptions(options);

        return factory;
    }

    // inbound -----------------------------------------------------------------------------------

    /**
     * 消息侦听回调
     */
    @Bean
    public MqttCallback inbound() {
        String[] topics = {"mqtt"};
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(IN_CLIENT_ID, mqttClientFactory(), topics);

        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.initialize();

        // 线程池
        adapter.setTaskScheduler(scheduler);

        // 消息转换器
        adapter.setConverter(new DefaultPahoMessageConverter());
        // 注释写操作超时，不知道具体指什么操作，先默认。应该是连接时候的超时，跟消息消费时间没关系
        adapter.setCompletionTimeout(MqttPahoMessageDrivenChannelAdapter.DEFAULT_COMPLETION_TIMEOUT);

        adapter.setQos(2);
        adapter.setOutputChannel(mqttInputChannel());

        return adapter;
    }

    /**
     * 注意函数名称，对应于 {@link ServiceActivator#inputChannel()}
     *
     * @return 消息通道
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * 消息处理
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return (message -> {
            Object topic = message.getHeaders().get("mqtt_receivedTopic");
            String payload = message.getPayload().toString();
            System.out.println("消息主题：" + topic);
            System.out.println("消息内容：" + payload);
        });
    }

    // outbound ----------------------------------------------------------------------------------

    /**
     * 注意函数名称，对应于 {@link ServiceActivator#outputChannel()}
     *
     * @return 消息通道
     */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * 消息发布
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler =
                new MqttPahoMessageHandler(OUT_CLIENT_ID, mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(2);
        messageHandler.setDefaultRetained(false);
        return messageHandler;
    }

}
